//用于处理ectd相关任务的CRUD
package worker

import (
	"context"
	"fmt"
	"gitee.com/Cookie_XiaoD/crontab/common"
	"github.com/coreos/etcd/clientv3"
	"github.com/coreos/etcd/mvcc/mvccpb"
	"time"
)

type JobMgr struct {
	client  *clientv3.Client
	lease   clientv3.Lease
	kv      clientv3.KV
	watcher clientv3.Watcher
}

var (
	G_JobMgr *JobMgr
)

//
func InitJobMgr() (err error) {
	//初始化任务管理器，配置etcd，获取连接以及租期操作
	var (
		cfg     clientv3.Config
		client  *clientv3.Client
		lease   clientv3.Lease
		kv      clientv3.KV
		watcher clientv3.Watcher
	)

	cfg = clientv3.Config{
		DialTimeout: time.Duration(G_config.EtcdDialTimeout) * time.Millisecond,
		Endpoints:   G_config.EtcdEndpoints,
	}

	if client, err = clientv3.New(cfg); err != nil {
		return
	}

	fmt.Println("Worker Etcd连接成功")

	kv = clientv3.NewKV(client)

	lease = clientv3.NewLease(client)

	watcher = clientv3.NewWatcher(client)

	G_JobMgr = &JobMgr{
		kv:      kv,
		client:  client,
		lease:   lease,
		watcher: watcher,
	}

	G_JobMgr.watchJobs()

	G_JobMgr.watchKills();

	return
}

func (jobMgr *JobMgr) watchJobs() (err error) {

	var (
		getResp            *clientv3.GetResponse
		kvpair             *mvccpb.KeyValue
		job                *common.Job
		watchStartRevision int64
		watchChan          clientv3.WatchChan
		jobEvent           *common.JobEvent
	)

	//获取现有的job，然后进行监听

	if getResp, err = jobMgr.kv.Get(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithPrefix()); err != nil {
		return
	}
	fmt.Println("初始化后获取任务信息")
	//获取成功后，反序列化为job
	for _, kvpair = range getResp.Kvs {
		if job, err = common.UnpackJob(kvpair.Value); err != nil {
			continue
		}
		jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)

		fmt.Println(*jobEvent)
		//TODO:推送事件给调度器
		G_scheduler.PushEventJob(jobEvent)
	}

	//从下一个版本开始监听
	go func() {
		watchStartRevision = getResp.Header.Revision + 1
		watchChan = jobMgr.watcher.Watch(context.TODO(), common.JOB_SAVE_DIR, clientv3.WithRev(watchStartRevision), clientv3.WithPrefix())

		for watchResp := range watchChan {
			for _, event := range watchResp.Events {
				switch event.Type {
				case mvccpb.PUT: //任务保存
					//对于任务保存，先反序列化得到任务对象，然后推送任务保存事件给调度器
					if job, err = common.UnpackJob(event.Kv.Value); err != nil {
						continue
					}
					fmt.Println("收到任务保存事件")
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_SAVE, job)
					//TODO:推送事件给调度器
					G_scheduler.PushEventJob(jobEvent)
				case mvccpb.DELETE: //任务删除
					//得到删除的key 格式是 /cron/jobs/任务名 ，而我们要得到的是任务名，所以需要截取
					//推送任务删除事件给调度器
					jobName := common.ExtractJobName(string(event.Kv.Key))
					job = &common.Job{
						Name: jobName,
					}
					fmt.Println("收到任务删除事件")
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_DELETE, job)
					//TODO:推送事件给调度器
					G_scheduler.PushEventJob(jobEvent)
				}
			}

		}
	}()

	return

}



func (jobMgr *JobMgr) watchKills() (err error) {

	var (
		job                *common.Job
		watchChan          clientv3.WatchChan
		jobEvent           *common.JobEvent
	)

	//从下一个版本开始监听
	go func() {
		watchChan = jobMgr.watcher.Watch(context.TODO(), common.JOB_KILL_DIR,clientv3.WithPrefix())

		for watchResp := range watchChan {
			for _, event := range watchResp.Events {
				switch event.Type {
				case mvccpb.PUT: //新的杀死任务事件，我们只关注PUT，因为任务强杀数据会自动过期
					fmt.Println("收到任务强杀事件")
					job=&common.Job{
						Name:common.ExtractKillJobName(string(event.Kv.Key)),
					}
					jobEvent = common.BuildJobEvent(common.JOB_EVENT_KILL, job)
					G_scheduler.PushEventJob(jobEvent)
				}
			}

		}
	}()

	return

}


//创建任务锁
func (jobMgr *JobMgr) CreateJobLock(info *common.JobExecuteInfo) (jobLock *JobLock) {
	jobLock = &JobLock{
		jobName: info.Job.Name,
		kv:      jobMgr.kv,
		lease:   jobMgr.lease,
	}
	return
}
